/**
 * Copyright 2007-2008. Chongqing First Information & Network Co., Ltd. All
 * rights reserved. <a>http://www.cqfirst.com.cn</a>
 */
package com.cqcis.uip.base.framework.base;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.ArrayList;
import java.util.Collection;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Set;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.mina.common.ConnectFuture;
import org.apache.mina.common.ExecutorThreadModel;
import org.apache.mina.common.IoConnector;
import org.apache.mina.common.IoFilter;
import org.apache.mina.common.IoFilterAdapter;
import org.apache.mina.common.IoFuture;
import org.apache.mina.common.IoFutureListener;
import org.apache.mina.common.IoSession;
import org.apache.mina.transport.socket.nio.SocketConnector;
import org.apache.mina.transport.socket.nio.SocketConnectorConfig;
import org.apache.mina.transport.socket.nio.SocketSessionImpl;

import com.cqcis.uip.base.framework.FrameworkConstants;
import com.cqcis.uip.base.framework.UIPServiceException;
import com.cqcis.uip.base.framework.base.adaptor.AbstractSocketService;
import com.cqcis.uip.base.framework.common.Role;
import com.cqcis.uip.base.framework.common.State;
import com.cqcis.uip.base.framework.common.Task;
import com.cqcis.uip.base.framework.core.RegisteredServiceContext;
import com.cqcis.uip.base.framework.core.TaskContextSessionStrategy;
import com.cqcis.uip.base.framework.filters.SocketFilterChainAdaptor;
import com.cqcis.uip.base.framework.handlers.common.AbstractClientHandler;
import com.cqcis.uip.base.framework.utils.LogServiceThreadPoolExecutor;
import com.cqcis.uip.base.framework.utils.TaskUtil;
import com.cqcis.uip.base.framework.utils.LogServiceThreadPoolExecutor.LogThreadPoolExecutor;

/**
 * Socket 客户端实现
 * 
 * @author huangmy
 * @date   2007-12-17
 */
public abstract class AbstractConnectorSupport extends AbstractSocketService {
    private static Log log = LogFactory.getLog(AbstractConnectorSupport.class);	    
	private IoConnector connector;
	private String address;        /*远程服务器IP地址*/
	private SocketConnectorConfig connectorConfig = new SocketConnectorConfig();	
	private SocketAddress socketAddress;  /*服务器地址*/	
	private ConnectFuture pendingConnectFuture; /*未决连接*/	
	private long timeout;
	private AbstractClientHandler handler;	
	private LogThreadPoolExecutor connectorTPE; // 监听线程池
	private TaskContextSessionStrategy<Object> taskContextSessionStrategy;
				
	/**
	 * 获取服务端地址
	 * @return
	 */
	private SocketAddress getSocketAddress() {
		if (socketAddress == null) {
			socketAddress = new InetSocketAddress(address, getPort());
		}
		
		return socketAddress;
	}
	
	public Role getRole() {		
		return Role.SOCKET_CLIENT;
	}
    
	@Override
	public boolean processResult(Task result) {
		boolean s = false;
		try {
			s = handler.processResult(result);
		} finally {
			afterProcessResult(result);
		}
		return s;
	}
	
	public void catchException(Task task, Throwable t) {
		try {
			handler.exceptionCaught(task.getSession(), t);
		} catch (Throwable e) {
			log.error("Cach Exception When catchException.", e);
		}
	}
	
	/**
	 * 建立连接
	 * 
	 * @param session
	 * @throws Exception
	 */
	public void clientNew(IoSession session) throws Exception {
		clientNew(session, getSocketAddress());
	}
	
	/**
	 * 建立连接
	 * @throws Exception
	 */
	public void clientNew() throws Exception {
		clientNew(null, getSocketAddress());
	}
	
	/**
	 * 建立连接
	 * 
	 * @param session
	 * @param address
	 * @throws Exception
	 */
	private void clientNew(IoSession session, SocketAddress address) throws Exception {	
		if (log.isDebugEnabled()) {
			log.debug("New A Client. remote address = " + address);
		}
		ConnectFuture future = null;
		if (session != null) {
			future = connector.connect(session, handler, connectorConfig);
		} else {
			future = connector.connect(address, handler, connectorConfig);
		}
		
		future.addListener(new IoFutureListener() {
			public void operationComplete(IoFuture ioFuture) {
				try {
					synchronized (handler) {
						IoSession session = ioFuture.getSession();
						Collection<IoSession> sessions = getConnectorSessions();
						if (sessions != null && sessions.size() > getClientNumberMax()) {
							if (session != null) {
								session.close();
							}
						}
					}				    			    
				} catch (Throwable t) {		
					if (taskContextSessionStrategy != null) {
						taskContextSessionStrategy.offerSession(t);
					}
										
					Collection<IoSession> sessions = getConnectorSessions();					
					if (sessions != null && sessions.size() == 0) {
						Task task = getTaskQueue().remove();
						if (task != null) {
							task.setMessage(FrameworkConstants.TASK_STATE_MSG_REMOTE_HOST_UNREACHED, t);
							task.setState(State.ERROR);
							
							dispatchTaskResult(task, false);
						}
						
						if (getTaskQueue().size() > 0) {
							connectionNew();
						}
					}
					
					log.error("ConnectFuture Exception: " + t.getCause(), t);
				}
			}
		});
		
		pendingConnectFuture = future;		
	}
	
	private IoFilter getCheckMaxClientFilter() {
		return new IoFilterAdapter() {
			@Override
			public void sessionCreated(NextFilter nextFilter, IoSession session)
					throws Exception {                				
				if (!isCloseOrCallNext(session)) {
					return;
				}
				
				super.sessionCreated(nextFilter, session);
			}
			
			@Override
			public void sessionOpened(NextFilter nextFilter, IoSession session)
					throws Exception {
				if (!isCloseOrCallNext(session)) {
					return;
				}
				
				super.sessionOpened(nextFilter, session);
			}
			
			private boolean isCloseOrCallNext(IoSession session) {
				Collection<IoSession> sessions = getConnectorSessions();
				if (sessions != null && sessions.size() > getClientNumberMax()) {
					if (session != null) {
						session.close();
						return false;
					}
				}
				return true;
			}
		};
	}
	
	@Override
	public void start(RegisteredServiceContext serviceContext) throws Exception {		
		super.start(serviceContext);
		
		addFilter("CheckMaxClientFilter", getCheckMaxClientFilter());
		
		int coreSize = threadPoolCoreSize/8;
		coreSize = coreSize > 2 ? coreSize : 2;
		coreSize = coreSize > 8 ? 8 : coreSize;
		connectorTPE = (LogThreadPoolExecutor)LogServiceThreadPoolExecutor.getLogExecutor(coreSize, coreSize * 4, 60, serviceContext.getLogService(), 
				getLogConfiguration(), "(C)"+getInterfaceName());
		
		int processors = coreSize - 2;
		processors = processors > 1 ? processors : 1;
		connector = new SocketConnector(processors, connectorTPE);	
		
		int connectTimeout = (int)timeout/1000;
		connectTimeout = (connectTimeout < 10) ? 10 : connectTimeout;
        connectorConfig.setConnectTimeout(connectTimeout);
		
		handler.registerTaskQueue(getTaskQueue()); // 处理任务
		handler.registerSendTaskQueue(getSendTaskQueue()); // 已处理任务
		handler.registerTaskResultQueue(getTaskResultQueue());// 任务结果
		handler.registerTimeout(timeout);
		handler.registerInterfaceId(getInterfaceId());
		handler.registerInterfaceLogService(getInterfaceLogService());
		handler.registerKeepAlive(isKeepAlive());
		handler.registerAdaptorRS(this);
		handler.registerPoolProps(threadPoolCoreSize, threadPoolMaxSize, threadPoolKeepAliveTime);
		handler.registerTaskWaitTimeConsumePool(getTaskWaitTimeConsumePool());
		handler.registerStreamRecord(isStreamRecord());
		handler.registerAskForSession(new Runnable() {
			public void run() {
				connectionNew();
			}
		});
		
		handler.start(serviceContext);
		
		doStart(); // 定制启动
	}
	
	@Override
	public void stop(RegisteredServiceContext serviceContext) throws Exception {
		// 关闭还未关闭的session
        Collection<IoSession> sessions = getManagedSessions();
        if (sessions != null) {        	
        	for (IoSession s : sessions) {
        		s.close().join();
        	}
        }
        
        sessions = getManagedSessions();
        if (sessions != null) {        	
        	for (IoSession s : sessions) {
        		s.close().join();
        		try {
        		    ((SocketSessionImpl)s).getChannel().close();
        		} catch (Throwable t) {        			
        		}
        	}
        }
        
        if (pendingConnectFuture != null) {
        	try {
				pendingConnectFuture.join();
				IoSession session = pendingConnectFuture.getSession();
				if (session != null) {
					session.close().join();
					try {
	        		    ((SocketSessionImpl)session).getChannel().close();
	        		} catch (Throwable t) {        			
	        		}
				}
			} catch (Throwable t) {
			}
        }
 
        connectorTPE.shutdownWait();    
        
        doStop(); // 定制停止
        
        super.stop(serviceContext);
		
		if (handler != null) {
			handler.stop(serviceContext);
		}
	}
	
	public AbstractClientHandler getHandler() {
		return this.handler;
	}
	
	@Override
	public Map<String, IoFilter> getFilters() {
		final LinkedHashMap<String, IoFilter> filters = new LinkedHashMap<String, IoFilter>();
		SocketFilterChainAdaptor sofilters = new SocketFilterChainAdaptor() {			
			@Override
			public void addFirst(String name, IoFilter filter) {
				filters.put(name, filter);
				super.addFirst(name, filter);
			}
			
			@Override
			public void addLast(String name, IoFilter filter) {
				filters.put(name, filter);
				super.addLast(name, filter);
			}
		};
		
		try {
			connectorConfig.getThreadModel().buildFilterChain(sofilters);
			this.connector.getFilterChainBuilder().buildFilterChain(sofilters);
			connectorConfig.getFilterChainBuilder().buildFilterChain(sofilters);			
		} catch (Exception e) {
			throw new UIPServiceException(e);
		}
        
		return filters;
	}
		
	/**
	 * 增加一个过滤器
	 * @param filterName
	 * @param filter
	 */
	protected void addFilter(String filterName, IoFilter filter) {
		if (filter != null) {				
			connectorConfig.getFilterChain().addLast(filterName, filter);
		}
	}
	
	@Override
	protected void setThreadPoolExecutorModelConfig(ExecutorThreadModel etm) {
		connectorConfig.setThreadModel(etm);
	}
	
	@Override
	public Collection<IoSession> getManagedSessions() {	
		Collection<IoSession> sessions = new ArrayList<IoSession>();
		if (connector != null) {
			Set<SocketAddress> mads = connector.getManagedServiceAddresses();
			if (mads != null) {				
				for (SocketAddress mad : mads) {
					Collection<IoSession> ss = connector.getManagedSessions(mad);
					if (ss != null) {
						sessions.addAll(ss);
					}
				}
			}
		}
		return sessions;
	}
	
	/**
	 * doStart details
	 * @throws Exception
	 */
	protected abstract void doStart() throws Exception;
	
	/**
	 * doStart details
	 * @throws Exception
	 */
	protected abstract void doStop() throws Exception;
	
	/**
	 * 新建一个session
	 */
	private void connectionNew() {			
		doConnectionNew();
	}
	
	/**
	 * 新建一个session
	 */
	private void doConnectionNew() {		
		try {
			if (!isNewConnect()) {
				return;
			}
			Collection<IoSession> sessions = getConnectorSessions();
			int count = getClientNumberMax() - sessions.size();
			count = (count > 0) ? count : 0;
			for (int i = 0; i < count; i ++) {				
			    clientNew();			
			}
		} catch (Exception e) {			
			log.error("Caugth Exception in trying to connect to remoteAddress: " +
					"address[" + address + "], port[" + getPort() + "]", e);
		}
	}
	
	@Override
	protected boolean doSchedule() {
		connectionNew();// 新建链接		
		return true;
	}
	
	@Override
	protected boolean doScheduleSync(Task task) {
		IoSession sessionBind = TaskUtil.getBindSession(task);
		if (sessionBind != null) {
			try {
				this.handler.doProcessTaskForSession(sessionBind, task);
			} catch (Throwable e) {
				task.setMessage(e.getMessage(), e);
				return false;
			}	
			return true;
		} else {
			boolean s = scheduleFor(task, false); 
			if (!s) {
				return false;
			}
		}
		
		return waitForSyncTaskFinishBlock(task);
	}
	
	private Collection<IoSession> getConnectorSessions() {
		return getManagedSessions();
	}
			
	/**
	 * 是否新建一个连接
	 * @return
	 */
	private boolean isNewConnect() {
		Collection<IoSession> sessions = getConnectorSessions();
		
		int sessionSize = sessions.size();
		int taskQueueSize = getTaskQueue().size();
		if (log.isDebugEnabled()) {
			log.debug("isNewConnect: clientNumberMax = " + getClientNumberMax() 
					+ ", sessionSize = " + sessionSize + ", taskQueueSize = " + taskQueueSize);
		}
        
		if (pendingConnectFuture != null && !(pendingConnectFuture.isReady())) {			
			return false;
		}
		
		if (sessionSize >= getClientNumberMax()) {			
			return false;
		}
		
		if (sessionSize > taskQueueSize) {
			return false;
		}
		
		return true;
	}
	
	public void setAddress(String remoteAddress) {
		this.address = remoteAddress;
	}
    
	public void setHandler(AbstractClientHandler handler) {
		this.handler = handler;
	}
	
	public void setTimeout(long timeout) {
		this.timeout = timeout;
	}

	public void registerTaskContextSessionStrategy(
			TaskContextSessionStrategy<Object> taskContextSessionStrategy) {
		this.taskContextSessionStrategy = taskContextSessionStrategy;
	}	
}
